package com.valten.factory;

import com.valten.model.RailWayModel;
import com.valten.repository.RailWayRepository;
import org.apache.camel.Exchange;
import org.apache.camel.Message;
import org.apache.camel.Processor;
import org.apache.camel.component.file.GenericFile;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;
import redis.clients.jedis.Jedis;
import redis.clients.jedis.Pipeline;

import java.io.*;
import java.util.ArrayList;
import java.util.Enumeration;
import java.util.List;
import java.util.concurrent.Callable;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.regex.Pattern;
import java.util.zip.ZipEntry;
import java.util.zip.ZipFile;
import java.util.zip.ZipInputStream;

/**
 * camel-ftp数据处理器
 *
 * @author huangyuanli
 * @className DataProcessor
 * @package com.valten.support
 * @date 2020/7/11 23:09
 **/
@Component
public class DataProcessor implements Processor {

    private static final Logger logger = LoggerFactory.getLogger(DataProcessor.class);
    private static final Pattern ALL_NUMBER_PATTERN = Pattern.compile("\\d+");

    @Value("${ftp.local-dir}")
    private String fileDir;
    @Value("${spring.redis.host}")
    private String host;
    @Value("${spring.redis.port}")
    private Integer port;
    @Value("${spring.redis.db}")
    private Integer db;

    @Autowired
    private RailWayRepository railWayRepository;

    @Override
    public void process(Exchange exchange) throws Exception {
        // 处理开始时间
        long startTime = System.currentTimeMillis();
        Message message = exchange.getIn();
        GenericFile<?> genericFile = (GenericFile<?>) message.getBody();
        String fileName = genericFile.getFileName();
        String filePath = fileDir + File.separator + fileName;
        // 解压文件
        readZip(filePath);
        // 数据合并处理
        handlerData(fileDir);

        // 清除解压后的文件和压缩包
        deleteFile(fileDir);
        // 处理结束时间
        long endTime = System.currentTimeMillis();
        System.out.println("耗时：" + (endTime - startTime));
    }

    /**
     * 删除文件夹下所有文件
     *
     * @author huangyuanli
     * @date 2020/7/11 23:10
     */
    private void deleteFile(String filePath) {
        // 获取目录名
        File rootfile = new File(filePath);
        File[] files = rootfile.listFiles();
        if (files != null) {
            // 遍历删除文件
            for (File file : files) {
                file.delete();
            }
        }
    }


    /**
     * 读取zip file中的文件内容
     *
     * @author huangyuanli
     * @date 2020/7/11 23:10
     */
    private void readZip(String filePath) throws Exception {
        ZipFile zipFile = new ZipFile(filePath);
        Enumeration<? extends ZipEntry> entries = zipFile.entries();
        InputStream inputStream = new BufferedInputStream(new FileInputStream(filePath));
        ZipInputStream zipInputStream = new ZipInputStream(inputStream);

        // 获取数据
        ZipEntry ze;
        // (ze=zipFile.getNextEntry())!=null 这样判断会导致ze.getSzie=-1
        while (entries.hasMoreElements()) {
            ze = entries.nextElement();
            if (ze.isDirectory()) {
                // 如果需要递归处理子目录，在此处实现
                File f = new File(fileDir + ze.getName());
                f.mkdir();
                continue;
            }
            logger.info("DetectorDataProcessor.readZip: Start reading {} in {}", ze.getName(), filePath);
            if (ze.getSize() > 0) {
                // 解压后文件名字
                String fileName = ze.getName();
                OutputStream os = new BufferedOutputStream(new FileOutputStream(new File(fileDir + File.separator + fileName)));
                InputStream is = new BufferedInputStream(zipFile.getInputStream(ze));
                byte[] buf = new byte[1024];
                int readLen;
                while ((readLen = is.read(buf, 0, 1024)) != -1) {
                    os.write(buf, 0, readLen);
                }
                is.close();
                os.close();
            }
        }
        zipInputStream.closeEntry();
        inputStream.close();
        zipFile.close();
    }

    /**
     * 数据合并处理 写入es
     *
     * @author huangyuanli
     * @date 2020/7/29 21:19
     */
    private void handlerData(String filePath) throws InterruptedException {
        List<String> saleInfos = getData(filePath);

        // 每1000条数据开启一条线程
        int threadSize = 100000;
        // 总数据条数
        int dataSize = saleInfos.size();
        // 线程数
        int threadNum = dataSize % threadSize == 0 ? dataSize / threadSize : dataSize / threadSize + 1;
        // 定义标记,过滤threadNum为整数
        boolean special = dataSize % threadSize == 0;
        // 线程池数量
        int poolSize = (2 * Runtime.getRuntime().availableProcessors()) + 1;
        // 创建一个线程池
        ExecutorService exec = Executors.newFixedThreadPool(poolSize);

        // 定义一个任务集合
        List<Callable<Integer>> tasks = new ArrayList<>();
        Callable<Integer> task;
        List<String> cutList;

        // 确定每条线程的数据
        for (int i = 0; i < threadNum; i++) {
            if (i == threadNum - 1) {
                if (special) {
                    break;
                }
                cutList = saleInfos.subList(threadSize * i, dataSize);
            } else {
                cutList = saleInfos.subList(threadSize * i, threadSize * (i + 1));
            }
            final List<String> listStr = cutList;
            task = () -> {
                Jedis jedis = new Jedis(host, port);
                jedis.select(db);

                List<RailWayModel> railWays = listStr.stream().collect(ArrayList::new, (list, item) -> {
                    RailWayModel railWay = new RailWayModel();
                    String[] saleItems = item.split(";");
                    // 乘车日期
                    String trainDate = saleItems[0];
                    // 车次
                    String boardTrainCode = saleItems[1];
                    // 发站
                    String fromStationName = saleItems[2];
                    // 到站
                    String toStationName = saleItems[3];
                    // 车厢
                    String coachNo = saleItems[4];
                    // 座位号
                    String seatNo = saleItems[5];
                    // 主键
                    String key = saleItems[6];

                    // 取票
                    if (!ALL_NUMBER_PATTERN.matcher(key).matches()) {
                        railWay.setIsTicket("1");
                    } else {
                        railWay.setIsTicket("0");
                    }

                    // 用户信息
                    boolean userInfoFlag = jedis.hexists("info_appendix", key);
                    if (userInfoFlag) {
                        String userInfoStr = jedis.hget("info_appendix", key);
                        String[] userItems = userInfoStr.split(";");
                        railWay.setIdKind(userItems[0]);
                        railWay.setIdNo(userItems[1]);
                        railWay.setName(userItems[2]);


                        // 取消
                        boolean cancelFlag = jedis.hexists("cancel", key);
                        if (cancelFlag) {
                            railWay.setIsCancel("1");
                        } else {
                            railWay.setIsCancel("0");
                        }

                        // 改签
                        boolean resignFlag = jedis.hexists("resign", key);
                        if (resignFlag) {
                            railWay.setIsResign("1");
                        } else {
                            railWay.setIsResign("0");
                        }

                        // 退票
                        boolean returnFlag = jedis.hexists("return", key);
                        if (returnFlag) {
                            railWay.setIsReturn("1");
                        } else {
                            railWay.setIsReturn("0");
                        }

                        // 证件号码
                        String idNo = railWay.getIdNo();
                        String id = idNo + boardTrainCode + trainDate;

                        RailWayModel result = railWayRepository.queryById(id);
                        // 如果已经有记录，表明既有购票信息又有取票信息
                        if (result != null) {
                            railWay.setIsTicket("1");
                        }

                        railWay.setId(id);
                        railWay.setTrainDate(trainDate);
                        railWay.setBoardTrainCode(boardTrainCode);
                        railWay.setFromStationName(fromStationName);
                        railWay.setToStationName(toStationName);
                        railWay.setCoachNo(coachNo);
                        railWay.setSeatNo(seatNo);

                        list.add(railWay);
                    }
                }, List::addAll);

                System.out.println(Thread.currentThread().getName() + "线程：" + railWays.size());
                if (!railWays.isEmpty()) {
                    railWayRepository.saveAll(railWays);
                }
                jedis.close();
                return 1;
            };
            // 这里提交的任务容器列表和返回的Future列表存在顺序对应的关系
            tasks.add(task);
        }

        exec.invokeAll(tasks);

        // 关闭线程池
        exec.shutdown();
    }

    /**
     * 数据合并处理
     *
     * @return 购票/取票信息
     * @author huangyuanli
     * @date 2020/7/12 21:40
     */
    private List<String> getData(String filePath) {
        // 购票/取票信息
        List<String> saleInfos = new ArrayList<>();
        File file = new File(filePath);
        // 列出文件
        File[] files = file.listFiles();
        if (files != null) {
            for (File f : files) {
                String fileName = f.getName();
                if (fileName.endsWith(".txt")) {
                    // 获取文件名（不带后缀）
                    String name = fileName.replaceAll("\\d", "").replaceAll("[.][^.]+$", "");
                    System.out.println(name);
                    switch (name) {
                        case "cancel":
                        case "resign":
                        case "return":
                            getRelationInfos(filePath, fileName);
                            break;
                        case "info_appendix":
                            getUserInfos(filePath, fileName);
                            break;
                        case "sale":
                            saleInfos = getSaleInfos(filePath, fileName);
                            break;
                        default:
                            break;
                    }
                }
            }
        }
        return saleInfos;
    }

    /**
     * 获取人员身份信息列表
     *
     * @author huangyuanli
     * @date 2020/7/12 13:31
     */
    public void getUserInfos(String filePath, String fileName) {
        Jedis jedis = new Jedis(host, port);
        jedis.select(db);
        Pipeline pipe = jedis.pipelined();
        try (FileInputStream fis = new FileInputStream(new File(filePath + File.separator + fileName));
             BufferedInputStream bis = new BufferedInputStream(fis);
             InputStreamReader isr = new InputStreamReader(bis);
             BufferedReader br = new BufferedReader(isr)) {

            // 获取文件名（不带后缀）
            String name = fileName.replaceAll("\\d", "").replaceAll("[.][^.]+$", "");
            String line;
            while ((line = br.readLine()) != null) {
                String[] info = line.split(";");
                String key = info[3];
                pipe.hset(name, key, line);
            }
            pipe.sync();
        } catch (IOException e) {
            jedis.close();
            pipe.close();
            e.printStackTrace();
        }
    }

    /**
     * 获取购票和取票信息列表
     *
     * @return 购票和取票信息
     * @author huangyuanli
     * @date 2020/7/12 13:31
     */
    public List<String> getSaleInfos(String filePath, String fileName) {
        List<String> saleInfos = new ArrayList<>();
        try (FileInputStream fis = new FileInputStream(new File(filePath + File.separator + fileName));
             BufferedInputStream bis = new BufferedInputStream(fis);
             InputStreamReader isr = new InputStreamReader(bis);
             BufferedReader br = new BufferedReader(isr)) {

            String line;
            while ((line = br.readLine()) != null) {
                saleInfos.add(line);
            }
            saleInfos.remove(0);
        } catch (IOException e) {
            e.printStackTrace();
        }
        return saleInfos;
    }

    /**
     * 获取取消、改签、退票信息列表
     *
     * @author huangyuanli
     * @date 2020/7/12 13:31
     */
    public void getRelationInfos(String filePath, String fileName) {
        Jedis jedis = new Jedis(host, port);
        jedis.select(db);
        Pipeline pipe = jedis.pipelined();
        try (FileInputStream fis = new FileInputStream(new File(filePath + File.separator + fileName));
             BufferedInputStream bis = new BufferedInputStream(fis);
             InputStreamReader isr = new InputStreamReader(bis);
             BufferedReader br = new BufferedReader(isr)) {

            // 获取文件名（不带后缀）
            String name = fileName.replaceAll("\\d", "").replaceAll("[.][^.]+$", "");
            String line;
            while ((line = br.readLine()) != null) {
                pipe.hset(name, line, line);
            }
            pipe.sync();
        } catch (IOException e) {
            jedis.close();
            pipe.close();
            e.printStackTrace();
        }
    }

}